package com.steve.summary.three_PC;

import java.util.List;
import java.util.concurrent.*;

/**
 * @Author: STEVE
 * @Description: 协调者
 * @since: 2025-05-21
 */
public class Coordinator {

    private final List<Participant> participants;
    private final ScheduledExecutorService timeoutScheduler = Executors.newScheduledThreadPool(1);

    public Coordinator(List<Participant> participants) {
        this.participants = participants;
    }

    /**
     * 执行3PC事务
     */
    public boolean execute3PC(String txId, long timeoutMs) {
        // Phase 1: CanCommit
        if (!phaseCanCommit(txId)) {
            return false;
        }

        // Phase 2: PreCommit
        if (!phasePreCommit(txId)) {
            sendAbortAll(txId);
            return false;
        }

        // Phase 3: DoCommit (with timeout monitor)
        return phaseDoCommit(txId, timeoutMs);
    }

    private boolean phaseCanCommit(String txId) {
        for (Participant p : participants) {
            if (!p.canCommit(txId)) {
                System.out.println("[Coordinator] Participant " + p.getParticipantId() + " CAN_COMMIT失败");
                return false;
            }
        }
        return true;
    }

    private boolean phasePreCommit(String txId) {
        for (Participant p : participants) {
            if (!p.preCommit(txId)) {
                System.out.println("[Coordinator] Participant " + p.getParticipantId() + " PRE_COMMIT失败");
                return false;
            }
        }
        return true;
    }

    private boolean phaseDoCommit(String txId, long timeoutMs) {
        // 设置超时监控（若协调者崩溃，参与者会超时自主提交）
        scheduleTimeoutCheck(txId, timeoutMs);
        // 发送DoCommit命令
        for (Participant p : participants) {
            if (!p.doCommit(txId)) {
                System.out.println("[Coordinator] Participant " + p.getParticipantId() + " DO_COMMIT失败");
                return false;
            }
        }
        return true;
    }

    private void sendAbortAll(String txId) {
        participants.forEach(p -> {
            if (!p.abort(txId)) {
                System.out.println("[Coordinator] 无法终止参与者: " + p.getParticipantId());
            }
        });
    }

    /**
     * 模拟超时检查（实际可用心跳机制替代）
     */
    private void scheduleTimeoutCheck(String txId, long timeoutMs) {
        timeoutScheduler.schedule(() -> {
            participants.forEach(p -> p.handleTimeout(txId));
        }, timeoutMs, TimeUnit.MILLISECONDS);
    }

}
